package com.xc.spark_dw.KuduSource

;


import java.time.LocalDateTime
import java.util.Calendar

import org.apache.kudu.spark.kudu.KuduContext
import org.apache.spark.sql.functions.col
import org.apache.spark.sql.types.DataTypes
import org.apache.spark.sql.{DataFrame, SparkSession}

object KuduExample {

  //  val sc = SparkSession.builder().appName("spark_test")
  //    .config("spark.testing.memory", "2147480000") //本地测试防止报内存不足,线上运行可取消
  //    .master("local").enableHiveSupport().getOrCreate()

  //  val sparkConf = new SparkConf().setAppName("kafkastreamming").setMaster("local[2]").set("spark.testing.memory","2147480000")
  //  val ssc = new StreamingContext(sparkConf, Seconds(2))
  //    ssc.checkpoint("checkpoint")
  val kuduMaster = "192.168.83.244:7051,192.168.83.82:7051,192.168.83.48:7051"
  val businessTableName = "impala::testbinlog.business"
  val baseUserTableName = "impala::testbinlog.base_user"
  val baseJobTableName = "impala::testbinlog.base_job"
  val baseUserJobTableName = "impala::testbinlog.base_user_job"
  val baseDepartTableName = "impala::testbinlog.base_department"
  val businessSuptTableName = "impala::testbinlog.business_supporter"
  val businessOnlineTableName = "impala::testbinlog.business_online_info"
  val onlineBusinessStatTableName = "impala::testbinlog.crm_result"
  val onlineBusinessDepartStatTableName = "impala::testbinlog.crm_depart_result"
  val baseUserOpt = getTableKuduOption(kuduMaster, baseUserTableName)
  val baseJobOpt = getTableKuduOption(kuduMaster, baseJobTableName)
  val baseUserJobOpt = getTableKuduOption(kuduMaster, baseUserJobTableName)
  val baseDepartOpt = getTableKuduOption(kuduMaster, baseDepartTableName)
  val businessOpt = getTableKuduOption(kuduMaster, businessTableName)
  val businessSuptOpt = getTableKuduOption(kuduMaster, businessSuptTableName)
  val businessOnlineOpt = getTableKuduOption(kuduMaster, businessOnlineTableName)
  //  val onlineBusinessStatOpt = getTableKuduOption(kuduMaster, onlineBusinessStatTableName)
  //  val onlineBusinessDepartStatOpt = getTableKuduOption(kuduMaster, onlineBusinessDepartStatTableName)


  def getTableKuduOption(kuduMaster: String, table: String): Map[String, String] = {
    Map("kudu.master" -> kuduMaster, "kudu.table" -> table)
  }

  def getDataFrameByKuduTable(sc: SparkSession, tableOptions: Map[String, String]): DataFrame = {
    sc.read.format("org.apache.kudu.spark.kudu").options(tableOptions).load().toDF()
  }


  def updateBusinessOnlineStat(sc: SparkSession, kuduContext: KuduContext) = {
    println("=====================================读入kudu数据========================================================")
    val sqlContext = sc.sqlContext
    val startTime = Calendar.getInstance.getTimeInMillis.toInt

    val baseUserDf = getDataFrameByKuduTable(sc, baseUserOpt)
    val baseJobDf = getDataFrameByKuduTable(sc, baseJobOpt)
    val baseUserJobDf = getDataFrameByKuduTable(sc, baseUserJobOpt)
    val baseDepartDf = getDataFrameByKuduTable(sc, baseDepartOpt)
    val businessDfSrc = getDataFrameByKuduTable(sc, businessOpt)
    val businessSuptDfSrc = getDataFrameByKuduTable(sc, businessSuptOpt)
    val businessOnlineDfSrc = getDataFrameByKuduTable(sc, businessOnlineOpt)

    val businessSuptDf = businessSuptDfSrc.filter(businessSuptDfSrc("del_flg") === 0).toDF() // 未删除
    val businessDf = businessDfSrc.filter(businessDfSrc("business_status") =!= 2) // 商机状态<>已关闭
      .filter(businessDfSrc("business_opportunity_type") === "dic-business-opportunity-type-002") // 商机类型 = 现金/定版金
    val businessOnlineDf = businessOnlineDfSrc
      .filter(businessOnlineDfSrc("year") === LocalDateTime.now().getYear) // 上刊日期
      .filter(businessOnlineDfSrc("month") === LocalDateTime.now().getMonth.getValue) // 上刊日期

    val baseUserJobDfSel = baseUserJobDf.select("user_id", "job_id", "depart_id")
      .withColumnRenamed("user_id", "job_user_id")
      .withColumnRenamed("depart_id", "job_depart_id")
    val businessDfSel = businessDf.select("business_id", "customer_id", "name", "belong_user_id", "online_date", "business_status", "business_opportunity_type")
    val baseUserDfSel = baseUserDf.select("user_id", "user_no", "name")
    val businessSuptDfSel = businessSuptDf.select("business_id", "achievement_radio", "del_flg")
      .withColumnRenamed("business_id", "business_id_s")
    val businessOnlineDfSel = businessOnlineDf.select("business_id", "amount", "year", "month")
      .withColumnRenamed("business_id", "business_id_o")

//    baseUserDf.show(10)

    baseDepartDf.cache()
    businessDf.cache()
    businessDfSel.cache()
    businessSuptDf.cache()
    businessSuptDfSel.cache()
    businessOnlineDf.cache()
    businessOnlineDfSel.cache()
    baseUserJobDf.cache()


    //      .filter(businessOnlineDfSel("year") === LocalDate.now().getYear)   // 上刊日期
    //    //      .filter(businessOnlineDfSel("month") === LocalDate.now().getMonth.getValue) // 上刊日期

    val departDfSel = baseDepartDf.withColumn("parent_id", baseDepartDf("parent_id").cast(DataTypes.IntegerType)).select("depart_id", "parent_id")
    departDfSel.cache()
    val departRootDfSel = departDfSel.filter(departDfSel("parent_id").isNull).select("depart_id", "parent_id")
    val departRootArray = departRootDfSel.select("depart_id").collect().map(_ (0)).toList.toArray
    val departSecondDf = departDfSel
      .where(col("parent_id").isin(departRootArray: _*))
      .select("depart_id", "parent_id").withColumnRenamed("depart_id", "depart_id_s")
      .withColumnRenamed("parent_id", "parent_id_s")
    val departSecondArray = departSecondDf.select("depart_id_s").collect().map(_ (0)).toList.toArray
    val departThirdDf = departDfSel
      .where(col("parent_id").isin(departSecondArray: _*))
      .select("depart_id", "parent_id").withColumnRenamed("depart_id", "depart_id_t")
      .withColumnRenamed("parent_id", "parent_id_t")
    val departThirdArray = departThirdDf.select("depart_id_t").collect().map(_ (0)).toList.toArray
    val departForthDf = departDfSel
      .where(col("parent_id").isin(departThirdArray: _*))
      .select("depart_id", "parent_id").withColumnRenamed("depart_id", "depart_id_f")
      .withColumnRenamed("parent_id", "parent_id_f")
    val departForthArray = departForthDf.select("depart_id_f").collect().map(_ (0)).toList.toArray
    val departFifthDf = departDfSel
      .where(col("parent_id").isin(departForthArray: _*))
      .select("depart_id", "parent_id").withColumnRenamed("depart_id", "depart_id_fi")
      .withColumnRenamed("parent_id", "parent_id_fi")
    val departFifthArray = departFifthDf.select("depart_id_fi").collect().map(_ (0)).toList.toArray
    val departSixthDf = departDfSel
      .where(col("parent_id").isin(departFifthArray: _*))
      .select("depart_id", "parent_id").withColumnRenamed("depart_id", "depart_id_six")
      .withColumnRenamed("parent_id", "parent_id_six")
    // depart_id tree
    val departUnionDf0 = departRootDfSel
      .join(departSecondDf, departRootDfSel("depart_id") === departSecondDf("parent_id_s"), "inner")
    val departUnionDf1 = departUnionDf0
      .join(departThirdDf, departUnionDf0("depart_id_s") === departThirdDf("parent_id_t"), "inner")
    val departUnionDf2 = departUnionDf1
      .join(departForthDf, departUnionDf1("depart_id_t") === departForthDf("parent_id_f"), "inner")
    val departUnionDf3 = departUnionDf2
      .join(departFifthDf, departUnionDf2("depart_id_f") === departFifthDf("parent_id_fi"), "inner")
    val departUnionDf = departUnionDf3
      .join(departSixthDf, departUnionDf3("depart_id_fi") === departSixthDf("parent_id_six"), "inner")

    //    departUnionDf.show()

    val departRootSixthDfSel = departUnionDf
      .select("depart_id", "depart_id_six")
      .withColumnRenamed("depart_id_six", "depart_business_id")

    val departRootFifthDfSel = departUnionDf
      .select("depart_id", "depart_id_fi")
      .withColumnRenamed("depart_id_fi", "depart_business_id")

    val departRootForthDfSel = departUnionDf
      .select("depart_id", "depart_id_f")
      .withColumnRenamed("depart_id_f", "depart_business_id")

    val departRootThirdDfSel = departUnionDf
      .select("depart_id", "depart_id_t")
      .withColumnRenamed("depart_id_t", "depart_business_id")
    val departRootSecondDfSel = departUnionDf
      .select("depart_id", "depart_id_s")
      .withColumnRenamed("depart_id_s", "depart_business_id")


    val departRootUnionDfSel = departRootDfSel
      .withColumn("depart_business_id", departRootDfSel("depart_id"))
      .select("depart_id", "depart_business_id")
      .union(departRootSecondDfSel)
      .union(departRootThirdDfSel)
      .union(departRootForthDfSel)
      .union(departRootFifthDfSel)
      .union(departRootSixthDfSel)
      .distinct()

    val departSecondDfSel = departUnionDf
      .select("depart_id_s")
      .withColumn("depart_business_id", departUnionDf("depart_id_s"))
      .withColumnRenamed("depart_id_s", "depart_id")
      .union(departUnionDf
        .select("depart_id_s", "depart_id_t")
        .withColumnRenamed("depart_id_s", "depart_id")
        .withColumnRenamed("depart_id_t", "depart_business_id"))
      .union(departUnionDf
        .select("depart_id_s", "depart_id_f")
        .withColumnRenamed("depart_id_s", "depart_id")
        .withColumnRenamed("depart_id_f", "depart_business_id")
      )
      .union(departUnionDf
        .select("depart_id_s", "depart_id_fi")
        .withColumnRenamed("depart_id_s", "depart_id")
        .withColumnRenamed("depart_id_fi", "depart_business_id")
      )
      .union(
        departUnionDf
          .select("depart_id_s", "depart_id_six")
          .withColumnRenamed("depart_id_s", "depart_id")
          .withColumnRenamed("depart_id_six", "depart_business_id")
      )
      .distinct()

    val departThirdDfSel = departUnionDf
      .select("depart_id_t")
      .withColumn("depart_business_id", departUnionDf("depart_id_t"))
      .withColumnRenamed("depart_id_t", "depart_id")
      .select("depart_id", "depart_business_id")
      .union(
        departUnionDf
          .select("depart_id_t", "depart_id_f")
          .withColumnRenamed("depart_id_t", "depart_id")
          .withColumnRenamed("depart_id_f", "depart_business_id")
          .select("depart_id", "depart_business_id")
      )
      .union(
        departUnionDf
          .select("depart_id_t", "depart_id_fi")
          .withColumnRenamed("depart_id_t", "depart_id")
          .withColumnRenamed("depart_id_fi", "depart_business_id")
          .select("depart_id", "depart_business_id")
      )
      .union(
        departUnionDf
          .select("depart_id_t", "depart_id_six")
          .withColumnRenamed("depart_id_t", "depart_id")
          .withColumnRenamed("depart_id_six", "depart_business_id")
          .select("depart_id", "depart_business_id")
      )
      .distinct()

    val departFourthDfSel = departUnionDf
      .select("depart_id_f")
      .withColumn("depart_business_id", departUnionDf("depart_id_f"))
      .withColumnRenamed("depart_id_f", "depart_id")
      .select("depart_id", "depart_business_id")
      .union(
        departUnionDf
          .select("depart_id_f", "depart_id_fi")
          .withColumnRenamed("depart_id_f", "depart_id")
          .withColumnRenamed("depart_id_fi", "depart_business_id"))
      .union(
        departUnionDf
          .select("depart_id_f", "depart_id_six")
          .withColumnRenamed("depart_id_f", "depart_id")
          .withColumnRenamed("depart_id_six", "depart_business_id")
          .select("depart_id", "depart_business_id"))
      .distinct()


    val departFifthDfSel = departUnionDf
      .select("depart_id_fi")
      .withColumn("depart_business_id", departUnionDf("depart_id_fi"))
      .withColumnRenamed("depart_id_fi", "depart_id")
      .union(departUnionDf
        .select("depart_id_fi", "depart_id_six")
        .withColumnRenamed("depart_id_fi", "depart_id")
        .withColumnRenamed("depart_id_six", "depart_business_id")
        .select("depart_id", "depart_business_id"))
      .distinct()

    val departSixthDfSel = departUnionDf
      .select("depart_id_six")
      .withColumn("depart_business_id", departUnionDf("depart_id_six"))
      .withColumnRenamed("depart_id_six", "depart_id")
      .select("depart_id", "depart_business_id")

    // 部门层级DF
    val unionDepartDfSel = departRootUnionDfSel
      .union(departSecondDfSel)
      .union(departThirdDfSel)
      .union(departFourthDfSel)
      .union(departFifthDfSel)
      .union(departSixthDfSel)
      .distinct()

    unionDepartDfSel.cache()

    //    val baseUserJobDfSel = baseUserJobDf.select("user_id", "job_id", "depart_id")
    //      .withColumnRenamed("user_id", "job_user_id")
    //      .withColumnRenamed("depart_id", "job_depart_id")
    //    val businessDfSel = businessDf.select("business_id", "customer_id", "name", "belong_user_id", "online_date", "business_status", "business_opportunity_type")
    //    val baseUserDfSel = baseUserDf.select("user_id", "user_no", "name")
    //    val businessSuptDfSel = businessSuptDf.select("business_id", "achievement_radio", "del_flg")
    //      .withColumnRenamed("business_id", "business_id_s")
    //    val businessOnlineDfSel = businessOnlineDf.select("business_id", "amount", "year", "month")
    //      .withColumnRenamed("business_id", "business_id_o")


    val userUnion0 = businessDfSel
      .join(baseUserDfSel, businessDfSel("belong_user_id") === baseUserDfSel("user_id"), "left")
    val userUnion1 = userUnion0
      .join(businessSuptDfSel, userUnion0("business_id") === businessSuptDfSel("business_id_s"), "left")
    val userUnion = userUnion1
      .join(businessOnlineDfSel, userUnion1("business_id") === businessOnlineDfSel("business_id_o"), "left")
    try{
      userUnion.cache()
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
        System.err.println("exception===>: ...")
      }
    }

    val departUnion0 = businessDfSel
      .join(baseUserDfSel, businessDfSel("belong_user_id") === baseUserDfSel("user_id"), "left")
    val departUnion1 = departUnion0
      .join(baseUserJobDfSel, departUnion0("user_id") === baseUserJobDfSel("job_user_id"), "inner")
    val departUnion2 = departUnion1
      .join(unionDepartDfSel, departUnion1("job_depart_id") === unionDepartDfSel("depart_business_id"), "inner")
    val departUnion3 = departUnion2
      .join(businessSuptDfSel, departUnion2("business_id") === businessSuptDfSel("business_id_s"), "left")
    val departUnion = departUnion3
      .join(businessOnlineDfSel, departUnion3("business_id") === businessOnlineDfSel("business_id_o"), "left")
      .distinct()

    try{
      departUnion.cache()
    } catch {
      case ex: Exception => {
        ex.printStackTrace()
        System.err.println("exception===>: ...")
      }
    }

    //    val userOnlineInfoDf = userUnion.filter(businessSuptDf("del_flg") === 0) // 未删除
    //      .filter(businessDf("business_status") =!= 2) // 商机状态<>已关闭
    //      .filter(businessDf("business_opportunity_type") === "dic-business-opportunity-type-002") // 商机类型 = 现金/定版金
    //      .distinct()
    //    //      .filter(businessOnlineDfSel("year") === LocalDate.now().getYear)   // 上刊日期
    //    //      .filter(businessOnlineDfSel("month") === LocalDate.now().getMonth.getValue) // 上刊日期
    //
    //    //    userOnlineInfoDf.show()
    //
    //    val departOnlineInfo = departUnion.filter(businessSuptDf("del_flg") === 0) // 未删除
    //      .filter(businessDf("business_status") =!= 2) // 商机状态<>已关闭
    //      .filter(businessDf("business_opportunity_type") === "dic-business-opportunity-type-002") // 商机类型 = 现金/定版金
    //      .distinct()

//    userUnion.show(10)
    import org.apache.spark.sql.functions._
    val userBusinessStatsInfo = userUnion
      .distinct()
      .dropDuplicates("business_id", "user_id")
      .withColumn("amount_", userUnion("amount") * userUnion("achievement_radio"))
      .groupBy("year", "month", "user_id")
      .agg(countDistinct("business_id"), countDistinct("customer_id"), sum("amount_"))
      .withColumnRenamed("count(DISTINCT business_id)", "online_business_count")
      .withColumnRenamed("count(DISTINCT customer_id)", "online_business_customer_count")
      .withColumnRenamed("sum(amount_)", "online_business_amount")
      .select("user_id", "year", "month", "online_business_count", "online_business_customer_count", "online_business_amount")
      .selectExpr("user_id", "year", "month", "online_business_count", "online_business_customer_count", "round(online_business_amount,2) as online_business_amount").toDF()


    //    userBusinessStatsInfo.show()

    //    import org.apache.spark.sql.
    //    val amount = departUnion.selectExpr("cast(amount as decimal(20,2)) as amount").toDF()
    //    val achievementRadio = departUnion.selectExpr("cast(achievement_radio as decimal(20,2)) as achievement_radio").toDF()

    //
    val departBusinessStatsInfo = departUnion
      .distinct()
      .dropDuplicates("business_id", "depart_id")
      .withColumn("amount_", departUnion("amount") * departUnion("achievement_radio"))
      .groupBy("year", "month", "depart_id")
      //      .agg(Map("business_id" -> "count", "customer_id" -> "count", "amount_" -> "sum"))
      .agg(countDistinct("business_id"), countDistinct("customer_id"), sum("amount_"))
      .withColumnRenamed("count(DISTINCT business_id)", "online_business_count")
      .withColumnRenamed("count(DISTINCT customer_id)", "online_business_customer_count")
      .withColumnRenamed("sum(amount_)", "online_business_amount")
      .select("depart_id", "year", "month", "online_business_count", "online_business_customer_count", "online_business_amount")
      .selectExpr("depart_id", "year", "month", "online_business_count", "online_business_customer_count", "round(online_business_amount,2) as online_business_amount").toDF()
//    val udfNull = udf((s: Any) => null)
    import org.apache.spark.sql.types.DataTypes
    //    val userBusinessStatsInfo_ = userBusinessStatsInfo.withColumn("depart_id", udfNull(col("user_id")).cast(DataTypes.IntegerType))
    val onlineBusinessUserResult = userBusinessStatsInfo
      .withColumn("user_id", userBusinessStatsInfo("user_id").cast(DataTypes.StringType))
      //        .withColumn("depart_id", userBusinessStatsInfo("depart_id").cast(DataTypes.IntegerType))
      .withColumn("year", userBusinessStatsInfo("year").cast(DataTypes.IntegerType))
      .withColumn("month", userBusinessStatsInfo("month").cast(DataTypes.IntegerType))
      .withColumn("online_business_amount", userBusinessStatsInfo("online_business_amount").cast(DataTypes.StringType))
      .withColumn("online_business_customer_count", userBusinessStatsInfo("online_business_customer_count").cast(DataTypes.IntegerType))
      .withColumn("online_business_count", userBusinessStatsInfo("online_business_count").cast(DataTypes.IntegerType))
      .na.drop(Array("year", "month"))
    //    val departBusinessStatsInfo_ = departBusinessStatsInfo.withColumn("user_id", udfNull(col("depart_id")).cast(DataTypes.IntegerType))
    val onlineBusinessDepartResult = departBusinessStatsInfo
      //        .withColumn("user_id", userBusinessStatsInfo("user_id").cast(DataTypes.StringType))
      .withColumn("depart_id", departBusinessStatsInfo("depart_id").cast(DataTypes.StringType))
      .withColumn("year", departBusinessStatsInfo("year").cast(DataTypes.IntegerType))
      .withColumn("month", departBusinessStatsInfo("month").cast(DataTypes.IntegerType))
      .withColumn("online_business_amount", departBusinessStatsInfo("online_business_amount").cast(DataTypes.StringType))
      .withColumn("online_business_customer_count", departBusinessStatsInfo("online_business_customer_count").cast(DataTypes.IntegerType))
      .withColumn("online_business_count", departBusinessStatsInfo("online_business_count").cast(DataTypes.IntegerType))
      .na.drop(Array("year", "month"))
    try {
      //      userBusinessStatsInfo.write.options(onlineBusinessStatOpt).mode("append").format("org.apache.kudu.spark.kudu").save


      println("===========================insert data=====================================")

      val insertStartTm = Calendar.getInstance.getTimeInMillis.toInt
      kuduContext.upsertRows(onlineBusinessUserResult, onlineBusinessStatTableName)
      val insertEndTm = Calendar.getInstance.getTimeInMillis.toInt
      val deltaTm = insertEndTm - insertStartTm
      println("===========================================================================")
      println(s"插入更新时间: $deltaTm")
      println("===========================================================================")

    } catch {
      case ex: Exception => ex.printStackTrace()
    }
    try {
      //      userBusinessStatsInfo.write.options(onlineBusinessStatOpt).mode("append").format("org.apache.kudu.spark.kudu").save

      println("===========================insert data=====================================")


      //      val kuduContext = new KuduContext(kuduMaster, sc.sparkContext)
      val insertStartTmDept = Calendar.getInstance.getTimeInMillis.toInt
      kuduContext.upsertRows(onlineBusinessDepartResult, onlineBusinessDepartStatTableName)
      kuduContext.upsertRows(onlineBusinessUserResult, onlineBusinessStatTableName)
      val insertEndTmDept = Calendar.getInstance.getTimeInMillis.toInt
      val deltaTmDept = insertEndTmDept - insertStartTmDept
      println("===========================================================================")
      println(s"插入更新时间: $deltaTmDept")
      println("===========================================================================")
    } catch {
      case ex: Exception => ex.printStackTrace()
    } finally {
      try {
        baseDepartDf.unpersist()
        businessDf.unpersist()
        businessDfSel.unpersist()
        businessSuptDf.unpersist()
        businessSuptDfSel.unpersist()
        businessOnlineDf.unpersist()
        businessOnlineDfSel.unpersist()
        baseUserJobDf.unpersist()
        departDfSel.unpersist()
        unionDepartDfSel.unpersist()
        userUnion.unpersist()
        departUnion.unpersist()
      }catch {
        case ex: Exception => ex.printStackTrace()
      }

    }
  }


  def updateBusinessOnlineStatBySql(sc: SparkSession) = {
    val businessReader = {
      sc.read.format("kudu").options(businessOpt).load()
    }
    businessReader.createTempView("business")
    sc.sqlContext.cacheTable("business")
    val datas = sc.sql("select * from business")
  }

  def main(args: Array[String]): Unit = {
    println("----------------------------------------------")
    val sc = SparkSession.builder().appName("spark_test")
      //      .config("spark.testing.memory", "2147480000") //本地测试防止报内存不足,线上运行可取消
      .config("spark.debug.maxToStringFields", "100")
      .master("local[8]").enableHiveSupport().getOrCreate()
    val kuduContext = new KuduContext(kuduMaster, sc.sparkContext)
    updateBusinessOnlineStat(sc, kuduContext)
  }


}
